package site.jlopen.service.imp;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;

import site.jlopen.assets.spring.KusciaApiInfoCache;
import site.jlopen.components.entity.GraphJson.GraphJsonNode;
import site.jlopen.components.entity.GraphJsonUtil;
import site.jlopen.components.entity.GraphListInfo;
import site.jlopen.domain.job.KusciaJob;
import site.jlopen.domain.job.KusciaJobResult;
import site.jlopen.domain.job.KusciaJobTask;
import site.jlopen.domain.job.event.KusciaJobTaskEvent;
import site.jlopen.domain.node.event.KusciaNodeEvent;
import site.jlopen.domain.project.KusciaProject;
import site.jlopen.domain.project.KusciaProjectFlow;
import site.jlopen.domain.project.KusciaProjectGraphParam;
import site.jlopen.domain.project.event.ProjectInviteNodeVO;
import site.jlopen.entity.BaseResponse;
import site.jlopen.entity.KusciaApiInfo;
import site.jlopen.entity.job.JobRunStatusVo;
import site.jlopen.entity.job.TaskLogVo;
import site.jlopen.entity.quartz.KusciaJobDataMap;
import site.jlopen.kuscia.model.job.TaskInfo;
import site.jlopen.kuscia.model.job.TaskPartie;
import site.jlopen.kuscia.service.job.AbstractJobManager;
import site.jlopen.mapper.data.KusciaDataGrantMapper;
import site.jlopen.mapper.job.KusciaJobMapper;
import site.jlopen.mapper.job.KusciaJobResultMapper;
import site.jlopen.mapper.job.KusciaJobTaskMapper;
import site.jlopen.mapper.node.KusciaNodeMapper;
import site.jlopen.mapper.project.KusciaProjectFlowMapper;
import site.jlopen.mapper.project.KusciaProjectGraphParamMapper;
import site.jlopen.mapper.project.KusciaProjectInviteMapper;
import site.jlopen.mapper.project.KusciaProjectMapper;
import site.jlopen.service.intf.project.JobResultService;
import site.jlopen.service.intf.project.JobTaskService;
import site.jlopen.service.intf.quartz.QuartzService;
import site.jlopen.utils.StringUtil;
import site.jlopen.utils.id.IdUtils;

@Service
public class JobTaskServiceImpl implements JobTaskService {

	@Autowired
	private KusciaProjectMapper projectMapper;
	@Autowired
	private KusciaProjectFlowMapper flowMapper;
	@Autowired
	private KusciaJobTaskMapper taskMapper;
	@Autowired
	private KusciaJobMapper jobMapper;
	@Autowired
	private KusciaNodeMapper nodeMapper;
	@Autowired
	private JobTaskConvert jobTaskConvert;
	@Autowired
	private AbstractJobManager jobManager;
	@Autowired
	private KusciaProjectInviteMapper inviteMapper;
	@Autowired
	private KusciaProjectGraphParamMapper graphParamMapper;
	//@Autowired private AsyncTaskService asyncTaskService;
	@Autowired 
	public JobResultService resultService;
	@Autowired 
	public KusciaJobResultMapper resultMapper;
	@Autowired
	private KusciaDataGrantMapper dataGrantMapper;
	@Autowired 
	private QuartzService quartzService;
	
	private KusciaApiInfo getApiInfo(Long ondeId) {
		KusciaNodeEvent node = nodeMapper.selectByKey(ondeId);
		KusciaApiInfo apiInfo = new KusciaApiInfo(node.getNodeProtocol(), node.getNodeGrpc())
				.ext(node.getNodeToken(), node.getNodeCert(), node.getNodeKey(), node.getNodeCa());
		return apiInfo;
	}
	
	private Long saveJob(Long projectId, Long flowId) {
		KusciaJob jobInfo = jobMapper.selectByInfo(new KusciaJob().setJobProjectId(projectId).setJobFlowId(flowId));
		KusciaProject project = projectMapper.selectByKey(projectId);
		if(null != jobInfo) {
			jobManager.deleteJob(getApiInfo(project.getProjectNodeId()), jobInfo.getJobId().toString());
			jobMapper.deleteByKey(jobInfo.getJobId());
			taskMapper.clearTask(jobInfo.getJobId());
			resultMapper.clearResultByProject(projectId, flowId);
		}
		KusciaJob job = new KusciaJob();
		job.setJobId(IdUtils.SnowFlakeId());
		job.setJobProjectId(projectId);
		job.setJobFlowId(flowId);
		job.setJobStatus("RUNNING");
		job.setJobDelete(0);
		job.setJobCreate(new Date());
		jobMapper.insert(job);
		return job.getJobId();
	}
	
	private KusciaJobTask createJobTask(Long projectId, Long flowId, String taskId, String graphNodeId, String graphNodeCode, String nodeInfoJson) {
		KusciaJobTask task = new KusciaJobTask();
		task.setTaskId(Long.valueOf(taskId));
		task.setTaskNodeId(graphNodeId);
		task.setTaskCode(graphNodeCode);
		task.setTaskJson(nodeInfoJson);
		return task;
	}
	
	private void saveJobTask(Long projectId, Long flowId, Long jobId, List<KusciaJobTask> tasks) {
		for (KusciaJobTask task : tasks) {
			//task.setTaskId(IdUtils.SnowFlakeId());
			task.setTaskJobId(jobId);
			task.setTaskDelete(0);
			task.setTaskCreate(new Date());
			if(null == task.getTaskStatus()) {
				task.setTaskStatus("PENDING");
			}
		}
		taskMapper.insertBatch(tasks);
	}
	
	@Override
	public BaseResponse<String> initJob(Long projectId, Long flowId) {
		List<KusciaJobTask> dbTasks = new ArrayList<KusciaJobTask>();
		Long dataSourceId = 0L;
		KusciaProject project = projectMapper.selectByKey(projectId);
		KusciaProjectFlow flow = flowMapper.selectByKey(flowId);
		List<ProjectInviteNodeVO> invites = inviteMapper.selectInviteNodes(projectId);
		List<TaskPartie> parties = new ArrayList<TaskPartie>();
		for (ProjectInviteNodeVO invite : invites) {
			parties.add(new TaskPartie(invite.getNodeCode(), ""));
			if("owner".equals(invite.getRole())) {
				dataSourceId = invite.getDatasourceId();
			}
		}
		if(StringUtil.isBlank(flow.getFlowJson())) {
			return new BaseResponse<String>().error("请完善流程组件");
		}
		GraphJsonUtil graphJsonUtil = new GraphJsonUtil(flow.getFlowJson());
		List<GraphListInfo> graphList = graphJsonUtil.getTasks();
		List<TaskInfo> tasks = new ArrayList<TaskInfo>();
		Map<String, String> newIdMap = new HashMap<String, String>();
		for (GraphListInfo graphInfo : graphList) {
			String newId = String.valueOf(IdUtils.SnowFlakeId());
			newIdMap.put(graphInfo.getId(), newId);
		}
		int grantCount = dataGrantMapper.selectAllGrant(flow.getFlowId());
		if(grantCount != invites.size()) {
			return new BaseResponse<String>().error("数据授权数量不匹配，请检查");
		}
		/*for (GraphListInfo graphInfo : graphList) {
			List<String> sources = graphInfo.getSources();
			List<String> newSources = new ArrayList<String>();
			if(null != sources && sources.size() > 0) {
				for (String source : sources) {
					newSources.add(newIdMap.get(source));
				}
				graphInfo.setSources(newSources);
			}
		}*/
		GraphJsonUtil graphUtil = new GraphJsonUtil(flow.getFlowJson());
		
		for (GraphListInfo graphInfo : graphList) {
			if("read_data".equals(graphInfo.getCode())) {
				GraphJsonNode graphNode = graphUtil.getNodeById(graphInfo.getId());
				KusciaProjectGraphParam paramInfo = graphParamMapper.selectByInfo(
						new KusciaProjectGraphParam().setParamFlowId(flow.getFlowId()).setParamGraphId(graphInfo.getId()));
				if(null == paramInfo) {
					return new BaseResponse<String>().error(graphNode.getData().getLabel() +  "节点参数未配置");
				}
				Map<String, Object> attrsMap = JSON.parseObject(paramInfo.getParamAttr(), new TypeReference<Map<String, Object>>() {});
				// 如果是数据读取则直接保存数据库
				KusciaJobTask rdtask = createJobTask(projectId, flowId, newIdMap.get(graphInfo.getId()), graphInfo.getId(), graphInfo.getCode(), JSON.toJSONString(graphInfo));
				rdtask.setTaskStatus("SUCCEEDED");
				dbTasks.add(rdtask);
				KusciaJobResult result = new KusciaJobResult();
				result.setResultProjectId(projectId);
				result.setResultFlowId(flowId);
				result.setResultTaskId(newIdMap.get(graphInfo.getId()));
				result.setResultDataSourceId(dataSourceId);
				result.setResultName("输入样本");
				result.setResultType("TABLE");
				result.setResultGraphId(graphInfo.getId());
				result.setResultGraphCode(graphInfo.getCode());
				result.setResultDomaindataId(attrsMap.get("datatable").toString());
				resultService.saveResult(result);
				continue;
			}
			List<String> dependencies = new ArrayList<String>();
			if(null != graphInfo.getSources() && graphInfo.getSources().size() > 0) {
				for (String sourceId : graphInfo.getSources()) {
					dependencies.add(sourceId);
				}
			}
			try {
				TaskInfo task = jobTaskConvert.toConvert(
						graphUtil,
						projectId, 
						flowId, 
						dataSourceId, 
						graphInfo.getId(), 
						graphInfo.getId(), 
						graphInfo.getId(),
						invites,
						dependencies,
						newIdMap
						);
				task.setTaskCode(graphInfo.getCode());
				task.setParties(parties);
				tasks.add(task);
				dbTasks.add(createJobTask(projectId, flowId, newIdMap.get(graphInfo.getId()), graphInfo.getId(), graphInfo.getCode(), JSON.toJSONString(task)));
			} catch (Exception e) {
				return new BaseResponse<String>().error("启动失败：" + e.getMessage());
			}
			
		}
		projectMapper.updateStatus(projectId, 1);
		Long jobId = saveJob(projectId, flowId);
		saveJobTask(projectId, flowId, jobId, dbTasks);
		KusciaNodeEvent node = nodeMapper.selectByKey(project.getProjectNodeId());
		boolean cJob = jobManager.createJob(getApiInfo(project.getProjectNodeId()), String.valueOf(jobId), node.getNodeCode(), tasks);
		if(cJob) {
			//asyncTaskService.startTask(getApiInfo(project.getProjectNodeId()), projectId, flowId, jobId);
			KusciaApiInfoCache.getInstance().addCache(project.getProjectNodeId(), getApiInfo(project.getProjectNodeId()));
			KusciaJobDataMap jobDataMap = new KusciaJobDataMap();
			jobDataMap.setProjectId(projectId);
			jobDataMap.setProjectName(project.getProjectName());
			jobDataMap.setFlowId(flowId);
			jobDataMap.setJobId(jobId);
			jobDataMap.setNodeId(project.getProjectNodeId());
			jobDataMap.setCallback(flow.getFlowCallback());
			quartzService.startKusciaJobListen(jobId, project.getProjectName() + "_" + flow.getFlowTitle(), jobDataMap);
		}
		return new BaseResponse<String>().success("运行启动运行中");
	}
	@Override
	public BaseResponse<String> startJob(Long projectId, Long flowId, String graphNodeId, Integer jobType) {
		if(jobType == 0) {
			return initJob(projectId, flowId);
		}
		if(jobType == 2) {
			KusciaJob jobInfo = jobMapper.selectByInfo(new KusciaJob().setJobProjectId(projectId).setJobFlowId(flowId));
			jobInfo.setJobStatus("READY");
			jobMapper.update(jobInfo);
			// return initJob(projectId, flowId);
			
			return new BaseResponse<String>().success("已恢复初始状态");
		}
		return new BaseResponse<String>().error("暂不支持执行单节点");
	}

	@Override
	public BaseResponse<String> stopJob(Long projectId, Long flowId) {
		KusciaProject project = projectMapper.selectByKey(projectId);
		KusciaJob jobInfo = jobMapper.selectByInfo(new KusciaJob().setJobProjectId(projectId).setJobFlowId(flowId));
		if(null != jobInfo) {
			jobInfo.setJobStatus("STOPED");
			jobMapper.updateSelective(jobInfo);
			KusciaNodeEvent node = nodeMapper.selectByKey(project.getProjectNodeId());
			jobManager.stopJob(getApiInfo(project.getProjectNodeId()), node.getNodeCode());
		}
		return new BaseResponse<String>().error("停止失败");
	}

	@Override
	public BaseResponse<List<JobRunStatusVo>> getTaskStatus(Long projectId, Long flowId) {
		KusciaJob jobInfo = jobMapper.selectByInfo(new KusciaJob().setJobProjectId(projectId).setJobFlowId(flowId));
		if(null == jobInfo) {
			return new BaseResponse<List<JobRunStatusVo>>().error("任务未运行");
		}
		List<JobRunStatusVo> list = taskMapper.selectTaskStatus(jobInfo.getJobId());
		return new BaseResponse<List<JobRunStatusVo>>().success("SUCCESS").setData(list);
	}

	@Override
	public BaseResponse<String> getJobStatus(Long projectId, Long flowId) {
		KusciaJob jobInfo = jobMapper.selectByInfo(new KusciaJob().setJobProjectId(projectId).setJobFlowId(flowId));
		if(null == jobInfo) {
			return new BaseResponse<String>().error("任务未运行");
		}
		return new BaseResponse<String>().success("任务查询成功", jobInfo.getJobStatus());
	}

	@Override
	public BaseResponse<List<String>> getTaskLog(Long projectId, Long flowId, String nodeId) {
		KusciaJob jobInfo = jobMapper.selectByInfo(new KusciaJob().setJobProjectId(projectId).setJobFlowId(flowId));
		if(null == jobInfo) {
			return new BaseResponse<List<String>>().error("任务未运行");
		}
		KusciaJobTaskEvent info = taskMapper.selectByInfo(new KusciaJobTask()
				.setTaskJobId(jobInfo.getJobId())
				.setTaskNodeId(nodeId)
				);
		if(null == info) {
			return new BaseResponse<List<String>>().error("查无该任务数据");
		}
		List<String> errs = new ArrayList<String>();
		if(StringUtil.isNotBlank(info.getTaskErrMsg())) {
			String[] errarr = info.getTaskErrMsg().split("\\\\n");
			errs = Arrays.asList(errarr);
		}
		return new BaseResponse<List<String>>().success("OK").setData(errs);
	}

	@Override
	public List<TaskLogVo> getTaskLogs(Long projectId, Long flowId) {
		return taskMapper.selectTaskLogs(projectId, flowId);
	}

	/*@Override
	public List<TaskResultDataVo> getResultDatas(Long projectId, Long flowId) {
		List<KusciaProjectGraphParamEvent> params = graphParamMapper.selectResultDatas(projectId, flowId);
		List<TaskResultDataVo> list = new ArrayList<TaskResultDataVo>();
		
		for (KusciaProjectGraphParamEvent param : params) {
			String outs = param.getParamOutputs();
			if(StringUtil.isNotBlank(outs)) {
				List<String> outss = JSON.parseArray(outs, String.class);
				for (String ddid : outss) {
					TaskResultDataVo vo = new TaskResultDataVo();
					vo.setCode(param.getParamGraphCode());
					vo.setDomainDataId(ddid);
					vo.setName(ddid);
					list.add(vo);
				}
			}
		}
		return list;
	}*/

}
